package com.fsql.controller;

import com.fsql.entity.JarFileInfo;
import com.fsql.entity.JarFileParamInfo;
import com.fsql.entity.JarJobParameterInfo;
import com.fsql.utils.FileUtils;
import com.fsql.utils.JobUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * flink jar包相关操作
 */
@RestController
@RequestMapping("/jars")
public class JarController {
    private static Logger LOG = LoggerFactory.getLogger(JobController.class);

    @Autowired
    private Environment env;

    @PostMapping("/run")
    public ResponseEntity<Map<String, Object>> submitsqljob(@RequestBody JarJobParameterInfo jobInfo) {
        Map<String, Object> map = new HashMap<>();
        if (StringUtils.isBlank(jobInfo.getFlinkConf())) {
            LOG.info("/jars/run flinkConf is empty, default value is {}", env.getProperty("flinksql.flinkconf"));
            jobInfo.setFlinkConf(env.getProperty("flinksql.flinkconf"));
        }
        map.put("jobId", JobUtils.submitStandaloneJob(jobInfo));
        return new ResponseEntity<>(map, HttpStatus.OK);
    }

    @PostMapping("/upload")
    public ResponseEntity<Object> uploadJar(JarFileParamInfo jarFile) {
        if (jarFile.getFile().isEmpty()) {
            return new ResponseEntity<>("上传文件空，请重新上传", HttpStatus.NO_CONTENT);
        }

        String fileName = jarFile.getFile().getOriginalFilename();
        String filePath = env.getProperty("file.uploadDir");
        String destFileName = filePath + File.separator + fileName;
        FileUtils.createFile(destFileName);
        File dest = new File(destFileName);
        try {
            jarFile.getFile().transferTo(dest);
            LOG.info("上传成功");

            JarFileInfo jarFileInfo = new JarFileInfo();
            jarFileInfo.setFileName(fileName);
            jarFileInfo.setFileSize(jarFile.getFile().getSize());
            jarFileInfo.setPath(dest.getPath());
            return new ResponseEntity<>(jarFileInfo, HttpStatus.OK);
        } catch (IOException e) {
            LOG.error(e.toString(), e);
        }
        return new ResponseEntity<>("上传失败", HttpStatus.INTERNAL_SERVER_ERROR);
    }
}
